package cn.sany.cterminal.config.mq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import cn.sany.cterminal.common.constants.CommonConst;
import cn.sany.cterminal.entity.mq.FaultInfoMsg;
import cn.sany.cterminal.entity.mq.MqMsg;
import cn.sany.cterminal.entity.mq.RealtimeInfoMsg;
import cn.sany.cterminal.service.impl.MqServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Slf4j
@Component
public class Consumer {
    @Autowired
    MqServiceImpl mqServiceImpl;
    /**
     * 消费者实体对象
     */
    private DefaultMQPushConsumer consumer;

    @Value("${rocketmq.consumer.groupName}")
    private String consumerGroup;
    @Value("${rocketmq.consumer.url}")
    private String url;
    @Value("${rocketmq.consumer.topic}")
    private String topic;

    /**
     * 通过构造函数 实例化对象
     */
    @PostConstruct
    public void initConsumer() throws MQClientException {

        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(url);
        //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //订阅主题和 标签（ * 代表所有标签)下信息
        consumer.subscribe(topic, "*");
        // //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs中只收集同一个topic，同一个tag，并且key相同的message
            // 会把不同的消息分别放置到不同的队列中
            try {
                for (Message msg : msgs) {

                    //消费者获取消息 这里只输出 不做后面逻辑处理
                    MqMsg mqMsg = JSONUtil.toBean(new String(msg.getBody()), MqMsg.class);
                    if(CommonConst.MqMsgType.REALTIME_INFO.equals(mqMsg.getDataType())){
                        mqServiceImpl.dealRealtimeInfo( JSONUtil.toBean((JSONObject)mqMsg.getData(),RealtimeInfoMsg.class));
                    }else if(CommonConst.MqMsgType.FAULT_INFO.equals(mqMsg.getDataType())){
                        mqServiceImpl.dealFaultInfo( JSONUtil.toBean((JSONObject)mqMsg.getData(),FaultInfoMsg.class));
                    }
                    log.info("Consumer get msg:{}",mqMsg.toString());
                }
            }catch (Exception e){
                log.error("数据解析异常：",e);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        System.out.println("消费者 启动成功=======");
    }
}